#include "ThreadPipe.h"
#include "RendererCommon.h"
#include <fcntl.h>
#include <sys/ioctl.h>
#include <ctime>
#include <memory>
#include <sys/select.h>
#include <unistd.h>

namespace simple_renderer{

ThreadPipe::ThreadPipe() : _stop_thread(false),
                           _signaled(false),
                           _msg_handler(nullptr),
                           _thread(nullptr),
                           _clear_tasks(false){

    _fd[0] = -1; _fd[1] = -1;
}

ThreadPipe::~ThreadPipe(){

}

void ThreadPipe::Process(){

    while(true){
        {
            std::unique_lock<std::mutex> lck(_pipe_mux);
            while(!_msg_list.empty()){
                Message msg = _msg_list.front();
                _msg_list.pop_front();
                _msg_handle_vec.push_back(msg);
            }
        }

        for(Message &msg : _msg_handle_vec){
            msg.handler(msg.data);
            if(msg.sync){
                LOGV("[ThreadPipe] notify start");
                std::unique_lock<std::mutex> lck1(*msg.cond_mux);
                (*msg.ready) = 1;
                msg.cond->notify_one();
                LOGV("[ThreadPipe] notify end");
            }
        }

        _msg_handle_vec.clear();
        if(_stop_thread && _msg_list.empty()){
            break;
        }
        Wait(200);
    }

    _msg_list.clear();
    _msg_handle_vec.clear();
    LOGV("[ThreadPipe] Process end");
}

void ThreadPipe::Post(std::function<void(MessageData *)> handler, MessageData *data){

    if(_stop_thread){
        return;
    }

    std::unique_lock<std::mutex> lck(_pipe_mux);
    Message msg;
    msg.data = data;
    msg.handler = handler;
    msg.sync = false;
    _msg_list.push_back(msg);
    WakeupPipe();
}

void ThreadPipe::PostSync(std::function<void(MessageData *)> handler, MessageData * data){

    if(_stop_thread){
        return;
    }

    Message msg;
    {
        std::unique_lock<std::mutex> lck(_pipe_mux);
        msg.data = data;
        msg.handler = handler;
        msg.sync = true;
        msg.ready = std::make_shared<uint8_t>(0);
        msg.cond_mux = std::make_shared<std::mutex>();
        msg.cond = std::make_shared<std::condition_variable>();

        _msg_list.push_back(msg);
    }
    WakeupPipe();

    std::unique_lock<std::mutex> lck1(*msg.cond_mux);
    while(!(*msg.ready)) {
        LOGV("[ThreadPipe] PostSync wait start");
        msg.cond->wait(lck1);
        LOGV("[ThreadPipe] PostSync wait end");
    }
    LOGV("[ThreadPipe] PostSync end");
}

void ThreadPipe::InsertHeadTask(Message msg){
    std::unique_lock<std::mutex> lck(_pipe_mux);
    _msg_list.push_front(msg);
    WakeupPipe();
}

void ThreadPipe::StartTask(){

    LOGV("[ThreadPipe] StartTask start");
    std::unique_lock<std::mutex> lck(_pipe_mux);
    OpenPipe();
    _stop_thread = false;
    _thread = std::move(std::unique_ptr<std::thread>(new std::thread(std::bind(&ThreadPipe::Process, this))));
    LOGV("[ThreadPipe] StartTask end");
}

void ThreadPipe::StopTask(){

    LOGV("[ThreadPipe] StopTask start");
    if(_thread) {
        _stop_thread = true;
        WakeupPipe();
        _thread->join();
        _thread = nullptr;
        ClosePipe();
    }
    LOGV("[ThreadPipe] StopTask end");
}

int ThreadPipe::OpenPipe(){

    if(pipe(_fd) < 0){
        LOGE("[ThreadPipe] OpenPipe init pipe error");
        return -1;
    }
    if (fcntl(_fd[0], F_SETFL, O_NONBLOCK) < 0) {
        LOGE("[ThreadPipe] OpenPipe set fd[0] nonblock error");
    }
    if (fcntl(_fd[1], F_SETFL, O_NONBLOCK) < 0) {
        LOGE("[ThreadPipe] OpenPipe set fd[1] nonblock error");
    }
    return 0;
}

int ThreadPipe::ClosePipe(){
    int ret;
    if(_fd[0] != -1) {
        ret = close(_fd[0]);
        LOGV("ClosePipe fd0 ret=%d", ret);
        _fd[0] = -1;
    }
    if(_fd[1 != -1]) {
        ret = close(_fd[1]);
        LOGV("ClosePipe fd1 ret=%d", ret);
        _fd[1] = -1;
    }
    return 0;
}

int ThreadPipe::WakeupPipe(){

    std::unique_lock<std::mutex> lck(_sig_mux);
    //write fd1  wakeup select
    if(!_signaled) {
        const uint8_t b[1] = {1};
        if (1 == write(_fd[1], b, sizeof(b))) {
            _signaled = true;
        }
    }
    return 0;
}

int ThreadPipe::ResetSignalPipe(){
    std::unique_lock<std::mutex> lck(_sig_mux);
    if(_signaled){
        uint8_t b[4] = {0};
        if(1 == read(_fd[0], b, sizeof(b))){
            _signaled = false;
        }
    }
    return 0;
}

void ThreadPipe::ClearTasks(){

    WakeupPipe();
}

WaitRes ThreadPipe::Wait(int wait_ms){

    struct timeval tv_wait = {0};
    tv_wait.tv_sec = wait_ms / 1000;
    tv_wait.tv_usec = (wait_ms%1000) * 1000;

    fd_set fds_read;
    FD_ZERO(&fds_read);
    fd_set fds_write;
    FD_ZERO(&fds_write);

    int fdmax = -1;
    if(_fd[0] > fdmax)
        fdmax = _fd[0];

    FD_SET(_fd[0], &fds_read);
    while(true){
        int n = select(fdmax + 1, &fds_read, &fds_write, nullptr, &tv_wait);
        if(n < 0){
            return kWait_Error;
        }else if(n == 0){
            return kWait_Timeout;
        }else{
            if(FD_ISSET(_fd[0], &fds_read)){
                ResetSignalPipe();
            }
            return kWait_Signaled;
        }
    }
}
}